package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object ComplexGroupCalculation {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf和SparkContext
    val conf = new SparkConf()
      .setAppName("Complex Group Calculation")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 1. 创建复杂的销售数据
    // 数据结构：(地区, 产品, 销售额, 销售日期)
    val salesData = List(
      ("北京", "手机", 2500.0, "2023-01-15"),
      ("上海", "电脑", 5000.0, "2023-02-20"),
      ("广州", "平板", 1800.0, "2023-03-10"),
      ("深圳", "手机", 3200.0, "2023-01-25"),
      ("北京", "电脑", 4800.0, "2023-02-05"),
      ("上海", "手机", 2800.0, "2023-03-15"),
      ("广州", "电脑", 4500.0, "2023-01-30"),
      ("深圳", "平板", 2000.0, "2023-02-15"),
      ("北京", "平板", 1500.0, "2023-03-05")
    )

    // Spark Stage划分的关键原则：
    // 1. 宽依赖操作会触发Stage的划分
    // 2. 窄依赖操作不会创建新的Stage
    // 3. 行动算子（collect）本身不会创建新的Stage，而是触发已有Stage的执行

    // Stage 0: 窄依赖阶段
    // parallelize、filter和map都是窄依赖操作
    // 这些操作不会触发shuffle，因此属于同一个Stage
    val salesRDD = sc.parallelize(salesData, 4)
      .filter(_._3 > 2000.0)  // 过滤高于2000元的销售记录
      .map(sale => (sale._1, sale._2, sale._3))  // 保留地区、产品、销售额

    // Stage 1: 宽依赖阶段（shuffle阶段）
    // groupByKey是宽依赖操作，会触发shuffle
    // 这是第二个Stage的开始
    val groupedByRegionProduct = salesRDD
      .map { case (region, product, amount) => 
        ((region, product), amount)  // 创建复合key
      }
      .groupByKey()  // 触发shuffle，结束前一个Stage

    // 为什么只有两个Stage？
    // 1. 第一个Stage（Stage 0）：
    //    - parallelize
    //    - filter
    //    - map
    //    这些都是窄依赖操作，不会触发shuffle

    // 2. 第二个Stage（Stage 1）：
    //    - 从map开始
    //    - groupByKey触发shuffle
    //    - 后续的mapValues和map都在这个Stage内
    //    - collect作为行动算子触发计算，但不会创建新Stage

    // collect的特殊性：
    // 1. 不会创建新的Stage
    // 2. 触发已有Stage的实际执行
    // 3. 将分布式计算的结果收集到Driver端

    val salesAnalysis = groupedByRegionProduct
      .mapValues(amounts => {
        val total = amounts.sum
        val count = amounts.size
        val avg = total / count
        val max = amounts.max
        val min = amounts.min
        (total, count, avg, max, min)
      })

    val repartitionedAnalysis = salesAnalysis
      .map { case ((region, product), (total, count, avg, max, min)) =>
        s"$region-$product: 总销售额=$total, 销售次数=$count, 平均销售额=$avg, 最高销售额=$max, 最低销售额=$min"
      }

    // collect触发Stage 1的实际计算
    // 但不会创建新的Stage
    println("销售分析结果：")
    repartitionedAnalysis.collect().foreach(println)

    // 暂停以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 